package com.example.demo.clq;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class ProducerConsumerTest {

    private static final Integer N_CPU = 4;

    private static final Long FAST = 1l;

    private static final Long SLOW = 1000l;

    /**
     * 待处理任务队列
     */
    public static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
    /**
     * 消费者线程个数
     */
    private static int threadCount = 5;
    /**
     * 线程池
     * LinkedBlockingQueue阻塞队列，默认为Integer.MAX_VALUE  -  无界队列
     */
    private static ExecutorService executorService = new ThreadPoolExecutor(threadCount + 1, threadCount + 1,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1000), new NameTreadFactory());

    private static volatile AtomicInteger stopProduce = new AtomicInteger(0);

    @PostConstruct
    private void init() {
        executorService.submit(new Producer());
        for (int i = 1; i <= threadCount; i++) {
            executorService.submit(new Consumer());
        }
    }

    /**
     * 生产者
     */
    static class Producer implements Runnable {

        @SneakyThrows
        @Override
        public void run() {
            while (true) {
                try {
                    if (stopProduce.intValue() < 10) {
                        queue.add("队列元素" + System.currentTimeMillis());
                        Thread.sleep(1000);
                        stopProduce.incrementAndGet();
                    }
                } catch (Exception e) {
                    log.error("线程发生异常！---------------------------------------------------------------------");
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 消费者
     */
    static class Consumer implements Runnable {

        @Override
        public void run() {
//            方法一：
//            while (true) {
//                try {
//                    if (!queue.isEmpty()) {
//                        String str = queue.poll();
//                        if (str != null && !str.isEmpty()) {
//                            log.error("线程:{},消费:{}", Thread.currentThread().getName(), str);
//                        }
//                    } else {
//                        Thread.sleep(100);
//                    }
//                } catch (Exception e) {
//                    log.error("线程发生异常！---------------------------------------------------------------------");
//                    e.printStackTrace();
//                }
//            }
            
//            方法二：
            AtomicBoolean atomicBoolean = new AtomicBoolean(true);
            ThreadLocalRandom random = ThreadLocalRandom.current();
            while (true) {
                try {
                    String str = this.getString(queue, atomicBoolean, random);
                    log.error("线程:{},消费:{}", Thread.currentThread().getName(), str);
                } catch (Exception e) {
                    log.error("线程发生异常！---------------------------------------------------------------------");
                    e.printStackTrace();
                }
            }
        }

        static String getString(final ConcurrentLinkedQueue<String> queue, final AtomicBoolean button,
                                final ThreadLocalRandom random) throws InterruptedException {
            String stamp = queue.poll();
            if (null != stamp) {
                log.error("first return stamp");
                return stamp;
            }
            int wait = N_CPU;

            // 感觉这个 retry: 有点秀的嫌疑
            retry:
            for (; ; ) {
                stamp = queue.poll();
                if (null != stamp) {
                    log.error("second return stamp");
                    return stamp;
                }
                log.error("Thread.sleep(FAST)");
                Thread.sleep(FAST);
                if (511 >= random.nextInt(1024)) {
                    --wait;
                    log.error("--wait, wait:{}", wait);
                }
                if (0 >= wait) {
                    log.error("进入retry, wait:{}", wait);
                    break retry;
                }
                if (button.get() == false) {
                    log.error("third return stamp");
                    return stamp;
                }
            }

            if (button.get() == false) {
                log.error("fourth return stamp");
                return stamp;
            }

            if (null == stamp) {
                log.error("Thread.sleep(SLOW)");
                Thread.sleep(SLOW);
            }
            log.error("fifth return stamp");
            return stamp;
        }
    }


    /**
     * 自定义的线程名称类
     * 定义线程池时使用
     */
    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger threadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, "my-thread-" + threadNum.getAndIncrement());
            log.error("{} has been created", thread.getName());
            return thread;
        }
    }

}


